package pers.twf.twfRpc.registry;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import pers.twf.twfRpc.config.RegistryConfig;
import pers.twf.twfRpc.model.ServiceMetaInfo;

import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

/**
 * zookeeper注册中心
 */
public class ZookeeperRegistry implements Registry{
//    curator提供的客户端
    private CuratorFramework client;
//    用于服务发现
    private ServiceDiscovery<ServiceMetaInfo> serviceDiscovery;
    /**
     * 根节点
     */
    private static final String ZK_ROOT_PATH = "/rpc/zk";


    @Override
    public void heartBeat() {
//        zookeeper中不需要心跳机制，临时节点连接断开时，会自动删除，不需要人工干预
    }

    /**
     * 消费端监听服务节点
     * @param serviceKey 服务节点Key
     */
    @Override
    public void watch(String serviceKey) {
        String watchKey = ZK_ROOT_PATH +"/"+ serviceKey;
        if (watchKeySet.add(watchKey)) {
            CuratorCache curatorCache = CuratorCache.build(client, watchKey);
            curatorCache.start();
            curatorCache.listenable().addListener(
                    CuratorCacheListener
                    .builder()
                    .forDeletes(childData -> registryServiceCache.clearCache(serviceKey))
                    .forChanges(((oldNode, node)-> registryServiceCache.clearCache(serviceKey)))
                            .build()
            );
        }

    }

    @Override
    public void init(RegistryConfig registryConfig) {
        client = CuratorFrameworkFactory
                .builder()
                .connectString(registryConfig.getAddress())
                .retryPolicy(new ExponentialBackoffRetry(Math.toIntExact(registryConfig.getTimeout()), 3))
                .build();

        serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMetaInfo.class)
                .client(client)
                .basePath(ZK_ROOT_PATH)
                .serializer(new JsonInstanceSerializer<>(ServiceMetaInfo.class))
                .build();

        try {
            client.start();
            serviceDiscovery.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        serviceDiscovery.registerService(buildServiceInstance(serviceMetaInfo));
        String registerKey = ZK_ROOT_PATH +"/"+ serviceMetaInfo.getServiceNodeKey();
        localRegisterNodeKeySet.add(registerKey);
    }

    private ServiceInstance<ServiceMetaInfo> buildServiceInstance(ServiceMetaInfo serviceMetaInfo) {
        String serviceAddress = serviceMetaInfo.getServiceHost()+":"+ serviceMetaInfo.getServicePort();
        try {
            return ServiceInstance.<ServiceMetaInfo>builder()
                    .id(serviceAddress)
                    .name(serviceMetaInfo.getServiceKey())
                    .address(serviceAddress)
                    .payload(serviceMetaInfo)
                    .build();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        try {
            serviceDiscovery.unregisterService(buildServiceInstance(serviceMetaInfo));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        String registerKey = ZK_ROOT_PATH +"/"+ serviceMetaInfo.getServiceNodeKey();
        localRegisterNodeKeySet.remove(registerKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        List<ServiceMetaInfo> cacheServiceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (cacheServiceMetaInfos!=null){
            return cacheServiceMetaInfos;
        }
        try {
            Collection<ServiceInstance<ServiceMetaInfo>> serviceInstances = serviceDiscovery.queryForInstances(serviceKey);
            watch(serviceKey);
            List<ServiceMetaInfo> serviceMetaInfoList = serviceInstances.stream()
                    .map(ServiceInstance::getPayload)
                    .collect(Collectors.toList());
            registryServiceCache.writeCache(serviceKey,serviceMetaInfoList);
            return serviceMetaInfoList;

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前节点下线");
//        都是临时节点，在Zookeeper中临时节点的生命周期依赖于客户端会话，会话结束时临时节点也会被删除
//        for (String key : localRegisterNodeKeySet) {
//            try {
//                client.delete().guaranteed().forPath(key);
//            } catch (Exception e) {
//                throw new RuntimeException(key+"节点下线失败");
//            }
//        }
        // 释放资源
        if (client != null) {
            client.close();
        }
    }
}
